package com.bootdo.common.rabbitmqconfig;


import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

//@Configuration
public class RabbitExchangeConfig {
    @Autowired
    private RabbitAdmin rabbitAdmin;

    @Autowired
    private ConnectionFactory connectionFactory;


    public RabbitExchangeConfig() {
    }


    //@Bean
    TopicExchange contractTopicExchangeDurable() {
        TopicExchange contractTopicExchange = new TopicExchange("com.exchange.topic",true,true);
        this.rabbitAdmin.declareExchange(contractTopicExchange);
        return contractTopicExchange;
    }

    //@Bean
    DirectExchange contractDirectExchange() {
        DirectExchange contractDirectExchange = new DirectExchange("com.exchange.direct");
        this.rabbitAdmin.declareExchange(contractDirectExchange);
        return contractDirectExchange;
    }




    /**
     * fanout队列
     * @return
     */
    @Bean
    FanoutExchange contractFanoutExchange() {
        FanoutExchange fanoutExchange = new FanoutExchange("com.timeout.fanout",true,true);
        this.rabbitAdmin.declareExchange(fanoutExchange);
        return fanoutExchange;
    }



    @Bean
    Queue timeOutQueue() {
        Queue queue = new Queue("com.haojiangbo.timewaitqueue", true);
        this.rabbitAdmin.declareQueue(queue);
        return queue;
    }


    /**
     * 绑定队列
     * @return
     */
    @Bean
    public Binding bindingTimeOutQueue() {
        return BindingBuilder.bind(timeOutQueue()).to(contractFanoutExchange());
    }



    /**
     * topIc队列
     * @return
     */
    @Bean
    TopicExchange haojiangboTopicExchange() {
        TopicExchange contractTopicExchange = new TopicExchange("com.haojiangbo.topic",true,true);
        this.rabbitAdmin.declareExchange(contractTopicExchange);
        return contractTopicExchange;
    }

    /**
     * 队列
     * @return
     */
    @Bean
    Queue haojiangboMasterQueue() {

        Map args = new HashMap();
        //设置生存时间 6秒
        args.put("x-message-ttl" , 5000);
        //超时后移送到哪个交换器
        args.put("x-dead-letter-exchange", "com.timeout.fanout");
        //超时移送到指定交换器的同时 制定key
        //args.put("x-dead-letter-routing-key","666ssss");
        Queue queue = new Queue("com.haojiangbo.masterqueue", true,false,false, args);
        this.rabbitAdmin.declareQueue(queue);
        return queue;
    }


    /**
     * 绑定队列
     * @return
     */
    @Bean
    public Binding binding2() {
        return BindingBuilder.bind(haojiangboMasterQueue()).to(haojiangboTopicExchange()).with("*.haojiangbo.*");
    }



    @Bean
    public SimpleMessageListenerContainer messageContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueues(timeOutQueue());
        container.setExposeListenerChannel(true);
        container.setMaxConcurrentConsumers(1);
        container.setConcurrentConsumers(1);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认
        container.setMessageListener(new ChannelAwareMessageListener() {
            @Override
            public void onMessage(Message message, Channel channel) throws Exception {
                byte[] body = message.getBody();
                System.err.println("延时队列等待消息中");
                System.err.println(new String(body,0,body.length));
                //确认消息
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //确认消息成功消费

            }
        });
        return container;
    }

    @Bean
    public SimpleMessageListenerContainer messageContainer2() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueues(haojiangboMasterQueue());
        container.setExposeListenerChannel(true);
        container.setMaxConcurrentConsumers(1);
        container.setConcurrentConsumers(1);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认
        container.setMessageListener(new ChannelAwareMessageListener() {
            @Override
            public void onMessage(Message message, Channel channel) throws Exception {
                byte[] body = message.getBody();
                System.err.println("正常队列得到消息" + new String(body));


                //值得注意的是  没有被消费的消息  或者被拒收的消息
                //只要这个方法接受到消息 没返回状态之前  延时队列都不会取出这个消息
                // Thread.sleep(1000*10);

                System.err.println("确认消息" + new String(body));

                //确认消息
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
/*
                System.err.println("拒绝确认" + new String(body));
                //取消确认
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false); //确认消息成功消费*/
            }
        });
        return container;
    }



}

